#define _GNU_SOURCE
#include <netdb.h>
#include <signal.h>
#include <errno.h>
#include <unistd.h>
#include <stdlib.h>
#include <sys/socket.h>
#include <setjmp.h>
#include <string.h>
#include <arpa/inet.h>
#include "main.h"
#include "log.h"
#include "task.h"
#include "buff.h"
#include "io.h"
#include "addr_ntop.h"
#include "tcp.h"


enum SocksVer {
    SOCKS_INVALID = 0x00,
    SOCKS4 = 0x04,
    SOCKS5 = 0x05,
};

enum Socks5Meth {
    S5_METH_NO_AUTH_REQ = 0x00,
    S5_METH_GSSAPI = 0x01,
    S5_METH_USERNAME_PASSWORD = 0x02,
    S5_METH_NO_ACCEPTABLE_METHODS = 0xFF,
};

enum Socks5Cmd {
    S5_CMD_CONNECT = 0x01,
    S5_CMD_BIND = 0x02,
    S5_CMD_UDP_ASSOCIATE = 0x03
};

enum Socks5Atyp {
    S5_ATYP_IPV4 = 0x01,
    S5_ATYP_DOMAINNAME = 0x03,
    S5_ATYP_IPV6 = 0x04
};

enum TcpReqStat {
    TR_STAT_REQ,
    TR_STAT_RSP_METH,
    TR_STAT_REQ_DETAILS,
    TR_STAT_RESOLV_ADDR,
    TR_STAT_CONNECT,
    TR_STAT_CONNECTING,
    TR_STAT_CONNECTED,
    TR_STAT_REPLY,
    TR_STAT_ERR
};

struct ReqAddr {
    enum Socks5Atyp atyp;
    union {
        struct in_addr in;
        struct in6_addr in6;
        char domain[256];
    };
    in_port_t port;
};

struct ConnAddr {
    union {
        struct sockaddr_in in;
        struct sockaddr_in6 in6;
    };
    int family;
    int socktype;
    int protocol;
    socklen_t addrlen;
};

enum Socks5Rep {
    S_REP_SUCCESSED = 0x00,
    S_REP_GENERAL_SOCKS_SRV = 0x01,
    S_REP_NOT_ALLOW = 0x02,
    S_REP_NET_UNREACH = 0x03,
    S_REP_HOST_UNREACH = 0x04,
    S_REP_CONN_REFUSED = 0x05,
    S_REP_TTL_EXPIRED = 0x06,
    S_REP_CMD_NOT_SUPPORT = 0x07,
    S_REP_ATYP_NOT_SUPPORT = 0x08,
};

struct TcpReqTask {
    struct task task;
    enum SocksVer ver; /* version */
    enum TcpReqStat stat;
    struct buff rbuf;
    struct buff wbuf;
    enum Socks5Cmd cmd;
    struct ReqAddr reqAddr; /* address info of request */
    struct ConnAddr connAddr; /* address of connect to server */
    struct TcpConnTask *srv;
    struct DomainResolv *resolv;
};

struct TcpConnTask {
    struct task task;
    struct buff buf;
    struct TcpConnTask *peer;
};

struct DomainResolv {
    int done;
    struct gaicb *reqgai; /* point to self, not allocate separate */
    struct sigevent sev; /* use signal to notification */
    struct TcpReqTask *task;
    struct DomainResolv *next;
} *domainResolvs;

#define SIG_DOMAIN_RESOLV_DONE SIGRTMIN

enum longjmpReturnInfo {
    JMP_RET_JMP_INIT = 0,
    JMP_RET_TASK_NORMAL = 1,
    JMP_RET_TASK_DELETED = -1,
    JMP_RET_TASK_PENDING_WRITE = 2,
};

jmp_buf jmpbuffer;

static struct TcpReqTask *tcpReqTaskCreate() {
    struct TcpReqTask *task = malloc(sizeof(*task));

    if (task) {
        task->task.fd = -1;
        task->task.handler = NULL;
        task->ver = SOCKS_INVALID;
        buffReset(&task->rbuf);
        buffReset(&task->wbuf);
        task->stat = TR_STAT_REQ;

        memset(&task->reqAddr, 0, sizeof(task->reqAddr));
        task->reqAddr.atyp = S5_ATYP_IPV4;
        task->srv = NULL;
        task->resolv = NULL;
    } else {
        errno = ENOMEM;
    }

    return task;
}

static void domainResolvFree(struct DomainResolv *priv, struct DomainResolv *curr) {
    if (curr->task)
        curr->task->resolv = NULL;
    if (curr->reqgai->ar_result)
        freeaddrinfo(curr->reqgai->ar_result);
    if (priv)
        priv->next = curr->next;
    else
        domainResolvs = curr->next;
    free(curr);
}

static void connectingTaskDel(const struct context *ctx, struct TcpConnTask *task);
static void reqTaskDel(const struct context *ctx, struct TcpReqTask *task) {
    LOG(LOG_DEBUG, "delete client requesting socket[%d]", task->task.fd);
    contextDeleteTask(ctx, &task->task);
    close(task->task.fd);
    task->task.fd = -1;
    if (task->resolv)
        task->resolv->task = NULL;
    if (task->srv && task->srv->task.fd >= 0)
        connectingTaskDel(ctx, task->srv);

    free(task);
}

static ssize_t writeOrSave(int fd, void *data, size_t nbytes, struct buff *savebuff) {
    ssize_t n, total = 0;

    if ((n = writeLowLevel(fd, data, nbytes)) != (ssize_t)nbytes) {
        if (n < 0) {
            if (errno == EAGAIN || errno == EWOULDBLOCK)
                n = 0;
            else
                return n;
        }
        total += n;
        n = buffPut(savebuff, data + total, nbytes - total);
        total += n;
    }

    return total;
}

static ssize_t reqTaskWrite(struct TcpReqTask *task, void *data, size_t nbytes) {
    return writeOrSave(task->task.fd, data, nbytes, &task->wbuf);
}

static void reqTaskFsm(const struct context *ctx, struct TcpReqTask *task);
static void reqTaskChangeStat(const struct context *ctx,
        struct TcpReqTask *task, enum TcpReqStat stat) {
    task->stat = stat;
    reqTaskFsm(ctx, task);
}

static void reqTaskToNextStat(const struct context *ctx, struct TcpReqTask *task) {
    reqTaskChangeStat(ctx, task, task->stat + 1);
}

static void reqTaskDie(const struct context *ctx, struct TcpReqTask *task) {
    reqTaskChangeStat(ctx, task, TR_STAT_ERR);
}

static size_t makeReplyMsg(struct TcpReqTask *task, enum Socks5Rep rep,
        uint8_t *buf, size_t buflen) {
    uint8_t *dst = buf;
    void *addrPtr;
    size_t addrBytes, outLen;

    switch (task->reqAddr.atyp) {
    case S5_ATYP_IPV4:
        addrPtr = &task->reqAddr.in;
        addrBytes = 4;
        outLen = 4 + addrBytes + 2;
        break;
    case S5_ATYP_IPV6:
        addrPtr = &task->reqAddr.in6;
        addrBytes = 16;
        outLen = 4 + addrBytes + 2;
        break;
    case S5_ATYP_DOMAINNAME:
        addrPtr = task->reqAddr.domain;
        addrBytes = strlen(addrPtr);
        outLen = 4 + 1 + addrBytes + 2;
        break;
    default:
        return 0;
    }

    if (buflen < outLen)
        return 0;

    *dst++ = task->ver;
    *dst++ = rep;
    *dst++ = 0x00;
    *dst++ = task->reqAddr.atyp;
    if (task->reqAddr.atyp == S5_ATYP_DOMAINNAME)
        *dst++ = addrBytes;

    memcpy(dst, addrPtr, addrBytes);
    *(in_port_t *)(dst + addrBytes) = task->reqAddr.port;

    return outLen;
}

static void replyAndDie(const struct context *ctx, struct TcpReqTask *task,
        enum Socks5Rep rep) {

    uint8_t buf[264];
    size_t msgLen;

    task->stat = TR_STAT_REPLY;
    msgLen = makeReplyMsg(task, rep, buf, sizeof(buf));
    LOG_DATA(LOG_DEBUG, buf, msgLen, "reply to client[%d]", task->task.fd);

    if (reqTaskWrite(task, buf, msgLen) < 0) {
        LOG(LOG_DEBUG, "write to client request socket[%d] error: %s",
                task->task.fd, strerror(errno));
        reqTaskDie(ctx, task);
    }

    if (task->wbuf.len == 0)
        reqTaskDie(ctx, task);

    longjmp(jmpbuffer, JMP_RET_TASK_PENDING_WRITE);
}

static void reqTaskRespMeth(const struct context *ctx, struct TcpReqTask *task,
        uint8_t meths[] __attribute((unused)), size_t methslen __attribute((unused))) {

    uint8_t methRsp[2];

    task->stat = TR_STAT_RSP_METH;

    // FIXME: check the methods is supported? and then chose one.

    methRsp[0] = task->ver;
    methRsp[1] = S5_METH_NO_AUTH_REQ;

    LOG_DATA(LOG_DEBUG, methRsp, sizeof(methRsp), "response METHOD to client[%d]", task->task.fd);

    if (reqTaskWrite(task, methRsp, sizeof(methRsp)) < 0) {
        LOG(LOG_DEBUG, "write to client request socket[%d] error: %s",
                        task->task.fd, strerror(errno));
        reqTaskDie(ctx, task);
    }

    if (task->wbuf.len == 0)
        reqTaskChangeStat(ctx, task, TR_STAT_REQ_DETAILS);
}

/**
 * return:
 *  RET_OK, RET_NEED_MORE, RET_NEED_DEL
 */
static void reqTaskRequest(const struct context *ctx, struct TcpReqTask *task) {
    uint8_t reqData[257];
    size_t methsLen;

    if (task->rbuf.len < 1)
        return;
    if (task->ver == SOCKS_INVALID) {
        buffPeek(&task->rbuf, &reqData[0], 0, 1);
        task->ver = reqData[0];
        if (task->ver != SOCKS5) {
            LOG(LOG_DEBUG, "unsupported socks version %d (task socket[%d])",
                    task->ver, task->task.fd);
            reqTaskDie(ctx, task);
        }
    }

    if (task->rbuf.len < 3)
        return;
    buffPeek(&task->rbuf, &reqData[1], 1, 1);
    methsLen = reqData[1];
    if (task->rbuf.len < 2 + methsLen)
        return;

    buffGet(&task->rbuf, reqData, 2 + methsLen);

    LOG_DATA(LOG_DEBUG, reqData, 2 + methsLen, "client[%d] request version/method", task->task.fd);

    reqTaskRespMeth(ctx, task, &reqData[2], methsLen);
}

static void reqTaskRequestDetails(const struct context *ctx, struct TcpReqTask *task) {
    uint8_t reqData[261];
    size_t dataLen;

    /* check VER */
    if (task->rbuf.len < 1)
        return; /* need more */
    buffPeek(&task->rbuf, &reqData[0], 0, 1);
    if (reqData[0] != task->ver) {
        LOG(LOG_DEBUG, "the socks version (%d) is inconsistent with the request (%d)",
                (int)reqData[0], task->ver);
        reqTaskDie(ctx, task);
    }

    /* check CMD */
    if (task->rbuf.len < 2)
        return; /* need more */
    buffPeek(&task->rbuf, &reqData[1], 1, 1);
    task->cmd = reqData[1];
    if (task->cmd != S5_CMD_CONNECT) {
        LOG(LOG_DEBUG, "unsupported command (%02hhX) (task socket[%d])",
                reqData[1], task->task.fd);
        replyAndDie(ctx, task, S_REP_CMD_NOT_SUPPORT);
    }

    /* check RSV and ATYP */
    if (task->rbuf.len < 4)
        return; /* need more */
    buffPeek(&task->rbuf, &reqData[2], 2, 2);
    if (reqData[2] != 0x00) {
        LOG(LOG_DEBUG, "invalid RESERVED field (%02hhX) of request details (task socket[%d])",
                 reqData[2], task->task.fd);
        reqTaskDie(ctx, task);
    }

    switch (reqData[3]) {
    case S5_ATYP_IPV4:
        dataLen = 6 + 4;
        if (task->rbuf.len < dataLen)
            return; /* need more */
        buffGet(&task->rbuf, reqData, dataLen);
        task->reqAddr.atyp = reqData[3];
        task->reqAddr.in.s_addr = *(in_addr_t *)&reqData[4];
        task->reqAddr.port = *(in_port_t *)&reqData[8];
        break;
    case S5_ATYP_IPV6:
        dataLen = 6 + 16;
        if (task->rbuf.len < dataLen)
            return; /* need more */
        buffGet(&task->rbuf, reqData, dataLen);
        task->reqAddr.atyp = reqData[3];
        memcpy(&task->reqAddr.in6, &reqData[4], 16);
        task->reqAddr.port = *(in_port_t *)&reqData[20];
        break;
    case S5_ATYP_DOMAINNAME:
        if (task->rbuf.len < 5)
            return; /* need more */
        buffPeek(&task->rbuf, &reqData[4], 4, 1);
        dataLen = 6 + 1 + reqData[4];
        if (task->rbuf.len < dataLen)
            return; /* need more */
        buffGet(&task->rbuf, reqData, dataLen);
        task->reqAddr.atyp = reqData[3];
        memcpy(task->reqAddr.domain, &reqData[5], reqData[4]);
        task->reqAddr.domain[reqData[4]] = '\0';
        task->reqAddr.port = *(in_port_t *)&reqData[5 + reqData[4]];
        break;
    default:
        LOG(LOG_DEBUG, "unsupported ATYPE (%02hhX) (task socket[%d])",
                 reqData[3], task->task.fd);
        replyAndDie(ctx, task, S_REP_ATYP_NOT_SUPPORT);
    }

    task->cmd = reqData[1];

#if DBG
    char msg[272];
    char addrstr[INET6_ADDRSTRLEN];
    switch (reqData[3]) {
    case S5_ATYP_IPV4:
        inet_ntop(AF_INET, &task->reqAddr.in, addrstr, sizeof(addrstr));
        snprintf(msg, sizeof(msg), "{[%s]:%hu}", addrstr, ntohs(task->reqAddr.port));
        break;
    case S5_ATYP_DOMAINNAME:
        snprintf(msg, sizeof(msg), "{%s:%hu}", task->reqAddr.domain, ntohs(task->reqAddr.port));
        break;
    case S5_ATYP_IPV6:
        inet_ntop(AF_INET6, &task->reqAddr.in6, addrstr, sizeof(addrstr));
        snprintf(msg, sizeof(msg), "{[%s]:%hu}", addrstr, ntohs(task->reqAddr.port));
        break;
    default:
        msg[0] = '\0';
    }
    LOG_DATA(LOG_DEBUG, reqData, dataLen, "client[%d] request details %s", task->task.fd, msg);
#endif

    reqTaskChangeStat(ctx, task, TR_STAT_RESOLV_ADDR);
}

static void addDomainResolv(const struct context *ctx, struct TcpReqTask *task) {
    int ret;
    struct DomainResolv *resolv;
    struct addrinfo *hint;
    char *domain;

    /* one element one malloc */
    resolv = malloc(sizeof(*resolv) + sizeof(*resolv->reqgai) +
            sizeof(*resolv->reqgai->ar_request) + strlen(task->reqAddr.domain) + 1);
    if (resolv == NULL) {
        LOG(LOG_DEBUG, "cannot allocate memory for getaddrinfo_a for client "
                "request socket[%d]", task->task.fd);
        replyAndDie(ctx, task, S_REP_GENERAL_SOCKS_SRV);
    }
    resolv->reqgai = (struct gaicb *) (resolv + 1);
    hint = (struct addrinfo *) (resolv->reqgai + 1);
    domain = (char *) (hint + 1);

    resolv->done = 0;

    strcpy(domain, task->reqAddr.domain);
    resolv->reqgai->ar_name = domain;
    resolv->reqgai->ar_service = NULL;
    memset(hint, 0, sizeof(*hint));
    hint->ai_flags = AI_CANONNAME;
    hint->ai_family = AF_UNSPEC;
    hint->ai_socktype = SOCK_STREAM;
    resolv->reqgai->ar_request = hint;
    resolv->reqgai->ar_result = NULL;

    resolv->sev.sigev_notify = SIGEV_SIGNAL;
    resolv->sev.sigev_signo = SIG_DOMAIN_RESOLV_DONE;
    resolv->sev.sigev_value.sival_ptr = resolv;

    resolv->task = task;

    ret = getaddrinfo_a(GAI_NOWAIT, &resolv->reqgai, 1, &resolv->sev);
    if (ret) {
        LOG(LOG_DEBUG, "getaddrinfo_a() for %s error: %s",
                task->reqAddr.domain, gai_strerror(ret));
        free(resolv);
        replyAndDie(ctx, task, S_REP_GENERAL_SOCKS_SRV);
    }

    task->resolv = resolv;

    if (domainResolvs)
        resolv->next = domainResolvs;
    else
        resolv->next = NULL;
    domainResolvs = resolv;
}

static void reqTaskResolvAddr(const struct context *ctx, struct TcpReqTask *task) {
    switch (task->reqAddr.atyp) {
    case S5_ATYP_IPV4:
        task->connAddr.in.sin_family = AF_INET;
        task->connAddr.in.sin_port = task->reqAddr.port;
        task->connAddr.in.sin_addr = task->reqAddr.in;
        task->connAddr.family = task->connAddr.in.sin_family;
        task->connAddr.addrlen = sizeof(struct sockaddr_in);
        task->connAddr.socktype = SOCK_STREAM;
        task->connAddr.protocol = IPPROTO_IP;
        reqTaskChangeStat(ctx, task, TR_STAT_CONNECT);
        break;
    case S5_ATYP_IPV6:
        task->connAddr.in6.sin6_family = AF_INET6;
        task->connAddr.in6.sin6_port = task->reqAddr.port;
        task->connAddr.in6.sin6_addr = task->reqAddr.in6;
        task->connAddr.family = task->connAddr.in6.sin6_family;
        task->connAddr.addrlen = sizeof(struct sockaddr_in6);
        task->connAddr.socktype = SOCK_STREAM;
        task->connAddr.protocol = IPPROTO_IP;
        reqTaskChangeStat(ctx, task, TR_STAT_CONNECT);
        break;
    case S5_ATYP_DOMAINNAME:
        addDomainResolv(ctx, task);
        break;
    default:
        replyAndDie(ctx, task, S_REP_ATYP_NOT_SUPPORT);
    }
}

static struct TcpConnTask *connTaskCreate() {
    struct TcpConnTask *task = malloc(sizeof(*task));

    if (task) {
        task->task.fd = -1;
        task->task.watchEvents.events = 0;
        task->task.watchEvents.data.ptr = NULL;
        task->task.handler = NULL;
        buffReset(&task->buf);
        task->peer = NULL;
    } else {
        errno = ENOMEM;
    }

    return task;
}

static void connectingTaskDel(const struct context *ctx, struct TcpConnTask *task) {
    LOG(LOG_DEBUG, "delete server connecting socket[%d]", task->task.fd);
    contextDeleteTask(ctx, &task->task);
    close(task->task.fd);
    task->task.fd = -1;
    if (task->peer && task->peer->task.fd >= 0)
        reqTaskDel(ctx, (struct TcpReqTask *)task->peer);

    free(task);
}

static void connectingTaskDie(const struct context *ctx, struct TcpConnTask *task) {
    replyAndDie(ctx, (struct TcpReqTask *)task->peer,
            S_REP_HOST_UNREACH);
}

static void tryConnect(const struct context *ctx, struct TcpReqTask *task) {
    enum Socks5Rep reply = S_REP_GENERAL_SOCKS_SRV;

    if (connectLowLevel(task->srv->task.fd, (struct sockaddr *)&task->connAddr,
            task->connAddr.addrlen) < 0) {
        switch(errno) {
        case EINPROGRESS:
            reqTaskChangeStat(ctx, task, TR_STAT_CONNECTING);
            return;
        case ECONNREFUSED:
            reply = S_REP_CONN_REFUSED;
            break;
        case ENETUNREACH:
            reply = S_REP_NET_UNREACH;
            break;
        case EHOSTUNREACH:
            reply = S_REP_NET_UNREACH;
            break;
        default:
            break;
        }
        LOG(LOG_DEBUG, "connect to request server failed: %s", strerror(errno));
        replyAndDie(ctx, task, reply);
    }
    reqTaskChangeStat(ctx, task, TR_STAT_CONNECTED);
}

static void reqTaskConnToServer(const struct context *ctx, struct TcpReqTask *task) {
    if ((task->srv = connTaskCreate()) == NULL) {
        LOG(LOG_WARNING, "create epoll task (for server connecting"
                " socket[%d]) failed: %s", task->srv->task.fd, strerror(errno));
        replyAndDie(ctx, task, S_REP_GENERAL_SOCKS_SRV);
    }
    task->srv->peer = (struct TcpConnTask *)task;

    task->srv->task.fd = socket(task->connAddr.family, task->connAddr.socktype,
            task->connAddr.protocol);
    if (task->srv->task.fd < 0) {
        LOG(LOG_DEBUG, "can not create socket: %s", strerror(errno));
        replyAndDie(ctx, task, S_REP_GENERAL_SOCKS_SRV);
    }
    setNonblocking(task->srv->task.fd);
    tryConnect(ctx, task);
}

static void connectEnd(const struct context *ctx, struct TcpConnTask *task) {
    int error;
    socklen_t len;
    enum Socks5Rep errReply = S_REP_GENERAL_SOCKS_SRV;

    len = sizeof(error);
    if (getsockopt(task->task.fd, SOL_SOCKET, SO_ERROR, &error, &len) == 0) {
        switch (error) {
        case 0: /* success */
            reqTaskChangeStat(ctx, (struct TcpReqTask *)task->peer, TR_STAT_CONNECTED);
            return;
        case EINTR:
            /* connect was interrupted, restart */
            tryConnect(ctx, (struct TcpReqTask *)task->peer);
            return;
        case ECONNREFUSED:
            errReply = S_REP_CONN_REFUSED;
            break;
        case ENETUNREACH:
            errReply = S_REP_NET_UNREACH;
            break;
        case EHOSTUNREACH:
            errReply = S_REP_NET_UNREACH;
            break;
        default:
            break;
        }
        LOG(LOG_DEBUG, "connect to request server failed: %s", strerror(errno));
    } else {
        LOG(LOG_DEBUG, "can not get server connecting socket[] SO_ERROR at "
                "SOL_SOCKET level: %s", strerror(errno));
    }

    replyAndDie(ctx, (struct TcpReqTask *)task->peer, errReply);
}

static enum EventHandlerRetVal connectingHandler(const struct context *ctx,
        const struct epoll_event *event) {
    struct TcpConnTask *task = event->data.ptr;

    switch (setjmp(jmpbuffer)) {
    case JMP_RET_JMP_INIT:
        break;
    case JMP_RET_TASK_DELETED:
        return RET_TASK_DELETED;
    default:
        return RET_NORMAL;
    }

    if (event->events & EPOLLOUT)
        connectEnd(ctx, task);

    if (event->events & (EPOLLRDHUP | EPOLLHUP | EPOLLERR))
        connectingTaskDie(ctx, task);

    return 0;
}

static void reqTaskWaitConnFinish(const struct context *ctx, struct TcpReqTask *task) {
    if (task->srv->task.handler == NULL) {
        task->srv->task.watchEvents.events = EPOLLOUT | EPOLLRDHUP;
        task->srv->task.handler = connectingHandler;
        task->srv->peer = (struct TcpConnTask *)task;
        if (contextAddTask(ctx, &task->srv->task) < 0) {
            LOG(LOG_WARNING, "server connecting socket[%d] cannot add to epoll "
                    "context: %s", task->srv->task.fd, strerror(errno));
            replyAndDie(ctx, task, S_REP_GENERAL_SOCKS_SRV);
        }
    }
}

static void estabTaskPairDie(const struct context *ctx, struct TcpConnTask *task) {
    struct TcpConnTask *peerTask = task->peer;

    LOG(LOG_DEBUG, "delete client-server-pair [%d<-->%d]", task->task.fd, peerTask->task.fd);

    contextDeleteTask(ctx, &peerTask->task);
    close(peerTask->task.fd);
    free(peerTask);

    contextDeleteTask(ctx, &task->task);
    close(task->task.fd);
    free(task);

    longjmp(jmpbuffer, JMP_RET_TASK_DELETED);
}

static enum EventHandlerRetVal establishedHandler(const struct context *ctx,
        const struct epoll_event *event) {
    struct TcpConnTask *task = event->data.ptr;
    struct TcpConnTask *peerTask = task->peer;

    switch (setjmp(jmpbuffer)) {
    case JMP_RET_JMP_INIT:
        break;
    case JMP_RET_TASK_PENDING_WRITE:
        goto DO_PENDING_WRITE;
    case JMP_RET_TASK_DELETED:
        return RET_TASK_DELETED;
    default:
        return RET_NORMAL;
    }

    if (event->events & (EPOLLHUP | EPOLLERR))
        estabTaskPairDie(ctx, task);

    if (event->events & EPOLLOUT) {
        if (buffWrite(&peerTask->buf, task->task.fd) < 0) {
            LOG(LOG_DEBUG, "client-server-pair write to %d error: %s",
                    task->task.fd, strerror(errno));
            estabTaskPairDie(ctx, task);
        }
        if (peerTask->buf.len == 0)
            contextTaskClearEvents(ctx, &task->task, EPOLLOUT);
    }

    if ((event->events & EPOLLIN) && buffRead(&task->buf, task->task.fd) < 0) {
        LOG(LOG_DEBUG, "client-server-pair read from %d error: %s",
                task->task.fd, strerror(errno));
        estabTaskPairDie(ctx, task);
    }

DO_PENDING_WRITE:
    if (task->buf.len && !(peerTask->task.watchEvents.events & EPOLLOUT)) {
        if (buffWrite(&task->buf, peerTask->task.fd) < 0) {
            LOG(LOG_DEBUG, "client-server-pair write to %d error: %s",
                    peerTask->task.fd, strerror(errno));
            estabTaskPairDie(ctx, peerTask);
        }
        if (task->buf.len)
            contextTaskSetEvents(ctx, &peerTask->task, EPOLLOUT);
    }

    if ((event->events & EPOLLRDHUP) && task->buf.len == 0)
        estabTaskPairDie(ctx, task);

    return RET_NORMAL;
}

static void reqTaskConnected(const struct context *ctx, struct TcpReqTask *task) {
    struct TcpConnTask *cli = NULL;
    struct TcpConnTask *srv;

    /* for client side */
    if ((cli = connTaskCreate()) == NULL) {
        LOG(LOG_WARNING, "create epoll task (for client success socket[%d]) "
                "failed: %s", task->task.fd, strerror(errno));
        replyAndDie(ctx, task, S_REP_GENERAL_SOCKS_SRV);
    }
    cli->task.fd = task->task.fd;
    cli->task.watchEvents.events = EPOLLIN | EPOLLRDHUP;
    cli->task.handler = establishedHandler;
    cli->peer = task->srv;
    if (contextTaskChange(ctx, (struct task *)cli) < 0) {
        LOG(LOG_WARNING, "client success socket[%d] cannot add to epoll "
                "context: %s", cli->task.fd, strerror(errno));
        free(cli);
        replyAndDie(ctx, task, S_REP_GENERAL_SOCKS_SRV);
    }

    /* for server side */
    srv = task->srv;
    srv->task.handler = establishedHandler;
    srv->peer = cli;
    if (contextTaskResetEvents(ctx, &srv->task, EPOLLIN | EPOLLRDHUP) < 0) {
        LOG(LOG_WARNING, "change epoll task (for server established"
                " socket[%d]) failed: %s", srv->task.fd, strerror(errno));
        free(cli);
        replyAndDie(ctx, task, S_REP_GENERAL_SOCKS_SRV);
    }

    /* reply success */
    uint8_t buf[264];
    size_t msgLen;
    msgLen = makeReplyMsg(task, S_REP_SUCCESSED, buf, sizeof(buf));
    free(task);
    LOG_DATA(LOG_DEBUG, buf, msgLen, "reply success to client[%d]", cli->task.fd);
    if (writeOrSave(cli->task.fd, buf, msgLen, &srv->buf) < 0) {
        LOG(LOG_DEBUG, "write to client request socket[%d] error: %s",
                cli->task.fd, strerror(errno));
        estabTaskPairDie(ctx, cli);
    }

    longjmp(jmpbuffer, JMP_RET_TASK_NORMAL);
}

static void reqTaskFsm(const struct context *ctx, struct TcpReqTask *task) {
    switch(task->stat) {
    case TR_STAT_REQ:
        reqTaskRequest(ctx, task);
        break;
    case TR_STAT_RSP_METH:
        break;
    case TR_STAT_REQ_DETAILS:
        reqTaskRequestDetails(ctx, task);
        break;
    case TR_STAT_RESOLV_ADDR:
        reqTaskResolvAddr(ctx, task);
        break;
    case TR_STAT_CONNECT:
        reqTaskConnToServer(ctx, task);
        break;
    case TR_STAT_CONNECTING:
        reqTaskWaitConnFinish(ctx, task);
        break;
    case TR_STAT_CONNECTED:
        reqTaskConnected(ctx, task);
        break;
    case TR_STAT_REPLY:
        break;
    case TR_STAT_ERR:
    default:
        reqTaskDel(ctx, task);
        longjmp(jmpbuffer, JMP_RET_TASK_DELETED);
    }
}

static void reqTaskDoWriteEvent(const struct context *ctx, struct TcpReqTask *task) {
    if (buffWrite(&task->rbuf, task->task.fd) < 0) {
        LOG(LOG_DEBUG, "write to client request socket[%d] error: %s",
                task->task.fd, strerror(errno));
        reqTaskDie(ctx, task);
    }

    if (task->wbuf.len == 0) {
        /* write finish, switch to next state */
        reqTaskToNextStat(ctx, task);
    }
}

static void reqTaskDoReadEvent(const struct context *ctx, struct TcpReqTask *task) {

    if (buffRead(&task->rbuf, task->task.fd) < 0) {
        LOG(LOG_DEBUG, "read from client request socket[%d] error: %s",
                task->task.fd, strerror(errno));
        reqTaskDie(ctx, task);
    }

    if (task->wbuf.len == 0 && task->rbuf.len >= 1)
        reqTaskFsm(ctx, task);
}

static enum EventHandlerRetVal tcpReqHandler(const struct context *ctx,
        const struct epoll_event *event) {
    struct TcpReqTask *task = event->data.ptr;

    switch (setjmp(jmpbuffer)) {
    case JMP_RET_JMP_INIT:
        break;
    case JMP_RET_TASK_PENDING_WRITE:
        goto DO_PENDING_WRITE;
    case JMP_RET_TASK_DELETED:
        return RET_TASK_DELETED;
    default:
        return RET_NORMAL;
    }

    if (event->events & (EPOLLRDHUP | EPOLLHUP | EPOLLERR)) {
        LOG(LOG_DEBUG, "client request socket[%d] die", task->task.fd);
        reqTaskDie(ctx, task);
    }

    if (event->events & EPOLLOUT)
        reqTaskDoWriteEvent(ctx, task);

    if (event->events & EPOLLIN)
        reqTaskDoReadEvent(ctx, task);

DO_PENDING_WRITE:
    if (task->task.watchEvents.events & EPOLLOUT) {
        if (task->wbuf.len == 0)
            contextTaskClearEvents(ctx, &task->task, EPOLLOUT);
    } else if (task->wbuf.len)
        contextTaskSetEvents(ctx, &task->task, EPOLLOUT);

    return RET_NORMAL;
}

static void domainResolvDoneSigAction(int sig __attribute((unused)),
        siginfo_t *val, void *ucp __attribute((unused))) {
    ((struct DomainResolv *)val->si_value.sival_ptr)->done = 1;
}

static int sigactions(void) {
    struct sigaction sa;

    sa.sa_sigaction = domainResolvDoneSigAction;
    sigemptyset(&sa.sa_mask);
    sa.sa_flags = SA_SIGINFO;
    if (sigaction(SIG_DOMAIN_RESOLV_DONE, &sa, NULL) < 0) {
        LOG(LOG_CRIT, "Can't catch DOMAIN_RESOLV_DONE_SIG : %s", strerror(errno));
        return -1;
    }

    return 0;
    return RET_NORMAL;
 }

static int createListenSocket(const char *host, const char *srv, int family) {
    int sockfd = -1, haveBindErr = 0;

    int err;
    struct addrinfo *ailist, *aip;
    struct addrinfo hint;

    memset(&hint, 0, sizeof(hint));
    hint.ai_flags = AI_PASSIVE;
    hint.ai_family = family;
    hint.ai_socktype = SOCK_STREAM;

    if ((err = getaddrinfo(host, srv, &hint, &ailist)) < 0) {
        LOG(LOG_CRIT, "getaddrinfo(%s, %s,&hint, &ailist) error: %s",
                host, srv, gai_strerror(err));
        return -1;
    }

    for (aip = ailist; aip; aip = ailist->ai_next) {
        if ((sockfd = socket(aip->ai_family, aip->ai_socktype, aip->ai_protocol)) < 0) {
            if (errno == EAFNOSUPPORT || errno == EPROTONOSUPPORT || errno == EPROTOTYPE)
                continue;
            LOG(LOG_CRIT, "Cannot create socket for %s,%s : %s",
                    host, srv, strerror(errno));
            break;
        }

        int reuse = 1;
        setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));

        if (bind(sockfd, aip->ai_addr, aip->ai_addrlen) == 0)
            break;
        haveBindErr = errno;

        close(sockfd);
        sockfd = -1;
    }

    if (sockfd < 0) {
        LOG(LOG_CRIT, "%s %s,%s failed: %s", haveBindErr ? "bind to" : "work on",
                host, srv, haveBindErr ? strerror(errno) : "No host found");
        goto FREEADDRINFO;
    }

    if (listen(sockfd, 0) < 0) {
        LOG(LOG_CRIT, "listen on %s,%s error: %s", host, srv, strerror(errno));
        close(sockfd);
        sockfd = -1;
        goto FREEADDRINFO;
    }

    char ipbuf[INET6_ADDRSTRLEN];
    if (sockaddrNtop(aip->ai_addr, ipbuf, sizeof(ipbuf)))
        host = ipbuf;
    LOG(LOG_INFO, "listen on %s,%s", host, srv);

FREEADDRINFO:
    freeaddrinfo(ailist);

    return sockfd;
}

static enum EventHandlerRetVal listenSocketHandler(const struct context *ctx,
        const struct epoll_event *event) {
    struct task *task = event->data.ptr;

    int connSock = accept(task->fd, NULL, NULL);
    if (connSock < 0) {
        switch (errno) {
        case EAGAIN:
#if EAGAIN != EWOULDBLOCK
        case EWOULDBLOCK:
#endif /* #if EAGAIN != EWOULDBLOCK */
        case EINTR:
        case ECONNABORTED:
            return RET_NORMAL;
        case ETIMEDOUT:
            LOG(LOG_NOTICE, "accept error: %s", strerror(errno));
            return RET_NORMAL;
        case EMFILE: case ENFILE:
        case ENOBUFS: case ENOMEM:
            LOG(LOG_WARNING, "accept error: %s", strerror(errno));
            return RET_NORMAL;
        default:
            LOG(LOG_ALERT, "accept error: %s", strerror(errno));
            return RET_SYS_ERR;
        }
    }

    struct TcpReqTask *spt;
    do {
        setNonblocking(connSock);
        if ((spt = tcpReqTaskCreate()) == NULL) {
            LOG(LOG_WARNING, "create epoll task (for client request socket[%d])"
                    " failed: %s", connSock, strerror(errno));
            break;
        }
        spt->task.fd = connSock;
        spt->task.watchEvents.events = EPOLLIN | EPOLLRDHUP;
        spt->task.handler = tcpReqHandler;
        if (contextAddTask(ctx, (struct task *)spt) < 0) {
            LOG(LOG_WARNING, "client request socket[%d] cannot add to epoll "
                    "context: %s", connSock, strerror(errno));
            break;
        }
        return RET_NORMAL;
    } while (0);

    close(connSock);
    if (spt)
        free(spt);
    return RET_NORMAL;
}

static int listenSocketAddToEpoll(struct context *ctx, int lsock) {
    struct task *task;
    if ((task = malloc(sizeof(struct task))) == NULL) {
        LOG(LOG_ALERT, "create epoll task (for listen socket[%d]) failed: "
                "Cannot allocate memory", lsock);
        return -1;
    }

    task->fd = lsock;
    task->watchEvents.events = EPOLLIN;
    task->handler = listenSocketHandler;
    if (contextAddTask(ctx, task) < 0) {
        LOG(LOG_ALERT, "listen socket[%d] cannot add to epoll context: %s",
                lsock, strerror(errno));
        free(task);
        return -1;
    }

    return 0;
}

static void domainResolvDone(const struct context *ctx,
        struct DomainResolv *priv, struct DomainResolv *curr) {
    int ret;
    struct addrinfo *aip;
    struct TcpReqTask *task;

    if ((task = curr->task) == NULL) {
        /* A solitary, can not successfully cancel at last, because it is in
         * progressing. now it done, we just do cancel */
        domainResolvFree(priv, curr);
        return;
    }

    if (setjmp(jmpbuffer))
        return;

    if ((ret = gai_error(curr->reqgai))) {
        LOG(LOG_NOTICE, "getaddrinfo_a \"%s\" for client request socket[%d] failed: %s",
                task->reqAddr.domain, task->task.fd, gai_strerror(ret));
        domainResolvFree(priv, curr);
        replyAndDie(ctx, task, S_REP_GENERAL_SOCKS_SRV);
    }

    for (aip = curr->reqgai->ar_result; aip; aip = aip->ai_next) {
        if (aip->ai_family == AF_INET || aip->ai_family == AF_INET6) {
            memcpy(&task->connAddr, aip->ai_addr, aip->ai_addrlen);
            task->connAddr.in.sin_port = task->reqAddr.port;
            task->connAddr.family = aip->ai_family;
            task->connAddr.addrlen = aip->ai_addrlen;
            task->connAddr.socktype = aip->ai_socktype;
            task->connAddr.protocol = aip->ai_protocol;
            break;
        }
    }

    domainResolvFree(priv, curr);

    if (aip == NULL) {
        LOG(LOG_INFO, "the domain %s not found", curr->reqgai->ar_name);
        replyAndDie(ctx, task, S_REP_GENERAL_SOCKS_SRV);
    }

    reqTaskChangeStat(ctx, task, TR_STAT_CONNECT);
}

int doTcpSocks() {
    struct context *ctx = NULL;
    int i, j;

    if (sigactions() < 0)
        return EXIT_FAILURE;

    if (G.listenIpv6)
        i = createListenSocket(G.listenIpv6, G.listenPort, AF_INET6);

    if (G.listenIpv4)
        j = createListenSocket(G.listenIpv4, G.listenPort, AF_INET);

    /* you maybe want fork */

    /* create epoll context */
    if ((ctx = contextCreate(-1)) == NULL) {
        LOG(LOG_ALERT, "epoll context create failed: %s", strerror(errno));
        if (i >= 0)
            close(i);
        if (j >= 0)
            close(j);
        return EXIT_FAILURE;
    }

    if (i >= 0 && listenSocketAddToEpoll(ctx, i) < 0) {
        close(i);
        i = -1;
    }

    if (j >= 0 && listenSocketAddToEpoll(ctx, j) < 0) {
        close(j);
        j = -1;
    }

    if (i < 0 && j < 0)
        return EXIT_FAILURE;

    while (1) {
        if ((j = epoll_wait(ctx->epollFd, ctx->events, ctx->size, -1)) < 0
                && errno != EINTR) {
            LOG(LOG_ALERT, "epoll_pwait error: %s", strerror(errno));
            break;
        }

        for (i = 0; i < j; ++i) {
            struct task *task = ctx->events[i].data.ptr;
            if (task) {
                switch (task->handler(ctx, &ctx->events[i])) {
                case RET_SYS_ERR:
                    exit(EXIT_FAILURE);
                case RET_TASK_DELETED:
                    i = j;
                    break;
                default:
                    break;
                }
            }
        }

        struct DomainResolv *curr, *priv, *next;
        for (priv = NULL, curr = domainResolvs; curr; curr = next) {
            next = curr->next; /* save the next, the curr may be deleted */
            if (curr->done)
                domainResolvDone(ctx, priv, curr);
            if (priv) {
                if (priv->next == curr)
                    priv = curr;
            } else {
                if (domainResolvs == curr)
                    priv = curr;
            }
        }
    }

    return EXIT_FAILURE;
}
